/**
 * AggActiveThread2.java
 * ************************** 
 * @date Apr 6, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package gossipServices.aggregation;

import gossipServices.aggregation.functions.AggregationUpdateFunction;
import gossipServices.basic.nodeDescriptors.ConcreteNodeDescriptor;
import gossipServices.basic.nodeDescriptors.NodeDescriptor;
import gossipServices.basic.view.PartialView;
import gossipServices.gossipMessages.AggregationMessage;
import gossipServices.gossipThread.ActiveGossipThreadInterface;
import gossipServices.network.NetworkServicesAPI;
import gossipServices.pss.PeerSamplingServiceAPI;

import java.net.InetAddress;
import java.net.UnknownHostException;

import messages.Message;

/**
 * Description: 
 * This class contains the instructions of the ActiveThread
 * of the Aggregation protocol. 
 * It extends the AggregationMiddleware class and implements
 * the ActiveGossipThreadInterface. 
 * The implementation of the method
 *  gossipThreadInstructions(), inherited from the base class,
 * calls the activeThreadInstructions function.
 *
 */
public class AggActiveThread extends AggregationMiddleware implements
		ActiveGossipThreadInterface {

	/**
	 * @param name
	 * @param cycleLenghtMsec
	 * @param view
	 * @param pss
	 * @param nia
	 * @param function
	 */
	public AggActiveThread(String name, long cycleLenghtMsec,
			PartialView view, PeerSamplingServiceAPI pss,
			NetworkServicesAPI nia, AggregationUpdateFunction function, double startingValue) {
		super(name, cycleLenghtMsec, view, pss, nia, function, startingValue);
	}

	@Override
	public void activeThreadInstructions() throws InterruptedException {
		NodeDescriptor q;
		NodeDescriptor me;
		Message msg;
		
		do{
			 q = pss.getPeer();
			 /* In the Aggregation a node can't extract itself!
			 (or we will lose values in the evaluations)
			 */
		}
		while(nia.getNodeName().equals(q.getNodeName()));
		
		try {
			me = new ConcreteNodeDescriptor(nia.getNodeName(), 
					InetAddress.getLocalHost(), new Integer(0));
			synchronized (estimationValue) {
				msg = new AggregationMessage(me, q, estimationValue,
						AggActiveThread.class,
						epochHandler.getEpoch(), 
						epochHandler.getRemainingCycles());				
			}
			nia.send(msg, q);
//			log.finer("sending msg: "+((AggregationMessage)msg)+" to "+q);
			AggregationMessage msgQ = (AggregationMessage) nia.receive(q);
			
			/*
			 * WARNING: 
			 *  option 1) make the receive(q) a blocking call
			 *    |-> may lead to a deadlock if messages are sent to crashed nodes
			 *   
			 *  option 2) verify the incoming message leaving unchanged the current value
			 */
			if(msgQ != null){				
//				log.finer("received: "+msgQ);
				
				switch(epochHandler.checkEpochs(msgQ, epochHandler)){
				case exchange:
//					log.info(currentThread().getName()+
//							" Epoch: "+epochHandler.getEpoch()
//							+" cycle: "+ epochHandler.getElapsedCycles()+
//							" exchanging with "+msgQ.getSender()+" before exchange: "+estimationValue);
					double sq = msgQ.getSingleValue().getValue();					
					exchange(sq);
					break;

				case sendCorrectEpoch:
					Message corrMsg = epochHandler.composeCorrectionMessage(me, msgQ.getSender(),
							AggActiveThread.class);
//					log.finer("send correction MSG: "+ corrMsg +" to "+msgQ.getSender() );
					nia.send(corrMsg, msgQ.getSender());
					break;

				case updateCurrentEpoch:
//					log.finer("updating current epoch from "+msgQ);
					epochHandler.updateCurrentEpoch(msgQ);
					break;
				
				default:

				}
			}
			
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}catch(AggregationRestartException e){
//			e.printStackTrace();
			restart(e);
		}
	}

	@Override
	public void gossipThreadInstructions() throws InterruptedException {
		activeThreadInstructions();
	}

}
